using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.NMS;
using Apache.NMS.Util;
using Apache.NMS.ActiveMQ;
using OPCAutomation;
using Newtonsoft.Json;

using System.Collections.Concurrent;
using System.Threading;
using System.Configuration;

using MySql.Data;
using MySql.Data.MySqlClient;

using CYQ.Data;

namespace PMC
{
    public class Convenyor
    {
        private OPCServer opcServer { get; set; }
        public OPCGroup ReadGroup { get; set; }

        private ConcurrentQueue<OpcTag> TagQueue { get; set; }

        private Dictionary<int, OpcTag> TagDictionary { get; set; }

        private volatile bool DataCollectWaitting=false;
        private Thread MessageCollectThread;
        private volatile bool DataCollectRunning;
        private AutoResetEvent DataCollectAutoResetEvent;

        
        public Convenyor()
        {
            TagQueue = new ConcurrentQueue<OpcTag>();

        }

        private List<OpcTag> LoadTags()
        {
            CYQ.Data.AppConfig.DB.DefaultConn = "mysql";

            using (MAction action = new MAction("t_tag"))
            {
                return action.Select().ToList<OpcTag>();
            }           
        }

        public void Start()
        {

            DataCollectRunning = true;
            MessageCollectThread = new Thread(new ThreadStart(MessageCollectTask));
            DataCollectAutoResetEvent = new AutoResetEvent(false);
            MessageCollectThread.Start();


            opcServer = new OPCAutomation.OPCServer();
            opcServer.Connect("KEPware.KEPServerEx.V4", "localhost");
            
            string GroupName = Guid.NewGuid().ToString();

            ReadGroup = opcServer.OPCGroups.Add(GroupName);
            ReadGroup.UpdateRate = 1000;
            ReadGroup.IsSubscribed = true;
            ReadGroup.IsActive = true;
            ReadGroup.DataChange += new DIOPCGroupEvent_DataChangeEventHandler(Group_DataChange);

            OPCItems Items = ReadGroup.OPCItems;
            TagDictionary = new Dictionary<int, OpcTag>();
            List<OpcTag> tags = LoadTags();
            if (tags != null)
            {
                int ClientHandle = 1;
                foreach (OpcTag tag in tags)
                {
                    Items.AddItem(tag.Name, ClientHandle);
                    tag.ClientHandle = ClientHandle;
                    TagDictionary.Add(ClientHandle, tag);
                    ClientHandle++;
                }
            }
        }

         

        void Group_DataChange(int TransactionID, int NumItems, ref Array ClientHandles, ref Array ItemValues, ref Array Qualities, ref Array TimeStamps)
        {
            try
            {

                for (int i = 1; i <= NumItems; i++)
                {
                    object Value=ItemValues.GetValue(i);
                    int ClientHandle = Convert.ToInt32(ClientHandles.GetValue(i));

                    if (TagDictionary.ContainsKey(ClientHandle)==false)
                    {
                        continue;
                    }

                    OpcTag tag= TagDictionary[ClientHandle];
                    object Quality = Qualities.GetValue(i);
                    if (Value == null) 
                    { 
                        continue; 
                    }
                    tag.Value = Value;                                       
                    TagQueue.Enqueue(tag);


                    if (DataCollectWaitting)
                    {
                        DataCollectAutoResetEvent.Set();
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
        }


        public void Stop()
        {
            opcServer.Disconnect();
            DataCollectRunning = false;
            DataCollectAutoResetEvent.Set();
            DataCollectWaitting = false;
            Thread.Sleep(1000);
            
        }

        private void MessageCollectTask()
        {
            while (DataCollectRunning)
            {
                while (TagQueue.Count > 0)
                {
                    OpcTag tag;
                    TagQueue.TryPeek(out tag);
                    if (tag == null)
                    {
                        TagQueue.TryDequeue(out tag);
                        continue;
                    }
                    string message = JsonConvert.SerializeObject(tag);
                    MQHelper.Send(message);
                    Console.WriteLine(message);
                    TagQueue.TryDequeue(out tag);
                }
                
                try
                {
                    DataCollectWaitting = true;
                    //等待信号释放
                    DataCollectAutoResetEvent.WaitOne();
                    //有信号释放，不用等待，下次循环继续等待执行信号
                    DataCollectWaitting = false;
                }
                catch (Exception e)
                {
                   
                }
                
            }
        }

       
    }
}
